package com.sprucetec.live.mcq.category;

import com.sprucetec.live.common.dto.business.LiveChatDto;
import com.sprucetec.live.mcq.McqBusinessMemoryMap;
import com.sprucetec.live.service.LiveChatService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 定时任务策略 注意：并发太高 线程池（chatCustomerThreadPool） 策略 会将部分消息丢弃
 */
@Slf4j
@Component("scheduledProcess")
@Scope("prototype")
public class ScheduledProcess implements LiveChatMegProcessCategory {

    private Lock lock = new ReentrantLock();


    private static final Integer scheduledPeriod = 10;// 10秒钟执行一次批量插入

    private volatile CountDownLatch countDownLatch = new CountDownLatch(0);


    private static final Integer batchInsertCursour = 10000; // 10000条 批量插


    private Integer cancelCondition = 12 * 60 * 60 / scheduledPeriod; // 如果12小时都没有数据要消费，默认当前直播关闭。进行关闭策略

    private Integer scheduleCurcor = 0; // 运行游标

    @Autowired
    private LiveChatService liveChatService;
    /**
     * 聊天数据存放数组
     */
    private List chatListContainer = new ArrayList();

    private Long videoId = null;

    @Override
    public void processChatMessage(LiveChatDto liveChatDto) {
        this.videoId = liveChatDto.getVideoId();
        try {
            lock.lock();
            ScheduledExecutorService scheduledExecutorService = McqBusinessMemoryMap.scheduledExecutorServiceMap.get(videoId);
            countDownLatch.await();
            log.info("消息加入到 chatListContainer" + liveChatDto.getContent());
            chatListContainer.add(liveChatDto);
            if (scheduledExecutorService == null) {
                scheduledExecutorService = Executors.newScheduledThreadPool(5);
                McqBusinessMemoryMap.scheduledExecutorServiceMap.put(videoId, scheduledExecutorService);
                scheduledExecutorService.scheduleAtFixedRate(new ResolveScheduledTask(), 1, scheduledPeriod, TimeUnit.SECONDS);
            }
        } catch (Exception ex) {
            log.error("error=>{}", ex);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 定时任务
     */
    class ResolveScheduledTask implements Runnable {
        @Override
        public void run() {
            countDownLatch = new CountDownLatch(1);
            try {
                if (chatListContainer.size() > batchInsertCursour) { // 超过一万条 分批次插入
                    List<Map<String, Object>> newList = null;//定义一个新的容器
                    int loopSize = chatListContainer.size() / batchInsertCursour;
                    for (int i = 0; i < loopSize; i++) {
                        if ((i + 1) == loopSize) {
                            //返回一个List集合的其中一部分视图。包含start，不包含end
                            newList = chatListContainer.subList((i * batchInsertCursour), chatListContainer.size());
                        } else {
                            newList = chatListContainer.subList((i * batchInsertCursour), ((i + 1) * batchInsertCursour));
                        }
                        batchInsertProxy(newList);
                    }
                } else {
                    batchInsertProxy(chatListContainer);
                }
                chatListContainer.clear();
                countDownLatch.countDown();
            } catch (Exception ex) {
                ex.printStackTrace();
                countDownLatch.countDown();
            }
        }
    }


    private void batchInsertProxy(List list) {
        if (list.size() == 0) {
            log.info("当前数组没有要插入数据库得聊天");
            if (scheduleCurcor < cancelCondition) {
                scheduleCurcor++;
            } else {
                this.close(); // 满足条件 清空相关数据，释放CPU资源
                log.info("消费聊天消息定时任务关闭");
            }
            return;
        }
        liveChatService.batchInsertChatByMq(list);
    }

    @Override
    public void close() {
        ScheduledExecutorService scheduledExecutorService = McqBusinessMemoryMap.scheduledExecutorServiceMap.get(videoId);
        if (scheduledExecutorService != null) {
            scheduledExecutorService.shutdown();
        }
        McqBusinessMemoryMap.scheduledExecutorServiceMap.remove(videoId);
    }
}
